package bolts;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

import common.TokenCount;
import common.UserTokenFrequency;

public class CalculateUserSimilarityBolt extends BaseBasicBolt {

    private double similarityThreshold;

    @Override
    public void prepare(Map stormConf, TopologyContext context) {
        super.prepare(stormConf, context);
        this.similarityThreshold = Double.valueOf((String) stormConf.get("simlarityThreshold"));
    }

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        final Map<String, List<TokenCount>> data = (Map<String, List<TokenCount>>) input.getValueByField("userTokenCounts");

        final List<UserTokenFrequency> users = new ArrayList<UserTokenFrequency>(data.size());

        for (final Entry<String, List<TokenCount>> entry : data.entrySet()) {
            users.add(new UserTokenFrequency(entry.getKey(), entry.getValue()));
        }

        for (int i = 0; i < users.size() - 1; i++) {
            for (int j = i + 1; j <= users.size() - 1; j++) {
                final double score = users.get(i).similarityScore(users.get(j));
                if (score >= similarityThreshold) {
                    collector.emit(new Values(new Date(), users.get(i).getUser(), users.get(j).getUser(), score));
                }
            }
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("time", "user1", "user2", "score"));
    }
}
